package com.bw.day0709;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.zb4.util.DateFormatUtil;
import com.bw.zb4.util.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Flink实时数据分流处理程序
 * 功能：从Kafka消费原始日志数据，进行数据清洗、新老用户识别和分流处理
 */


public class dwd_gd3 {
    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1（生产环境应根据实际情况调整）
        env.setParallelism(1);

        // 2. 从Kafka读取原始数据
        String sourceTopic = "topic_gd3_ods";       // 原始数据主题
        String groupId = "gd3_dwd_group";          // 消费者组ID
        // 使用自定义工具类获取Kafka消费者
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(sourceTopic, groupId));

        // 3. 数据清洗和转换
        SingleOutputStreamOperator<JSONObject> mappedStream = source.map(jsonStr -> {
                    try {
                        // 将JSON字符串解析为JSONObject
                        return JSON.parseObject(jsonStr);
                    } catch (Exception e) {
                        // 打印解析失败的JSON字符串
                        System.err.println("JSON解析失败: " + jsonStr);
                        // 返回空JSONObject
                        return new JSONObject();
                    }
                })
                // 过滤掉空JSONObject
                .filter(jsonObj -> jsonObj != null && !jsonObj.isEmpty());

        // 4. 修复keyBy操作 - 确保使用有效的key字段
        KeyedStream<JSONObject, String> keyedStream = mappedStream
                .keyBy(r -> {
                    // 获取key字段值，如果为null则使用默认值"default_key"
                    String keyValue = r.getString("key");
                    return keyValue != null ? keyValue : "default_key";
                });

        // 5. 新老用户识别处理
        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    // 定义状态变量，存储首次访问日期
                    ValueState<String> firstViewDtState;

                    @Override
                    public void open(Configuration param) throws Exception {
                        super.open(param);
                        // 初始化状态变量
                        firstViewDtState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("firstViewDtState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            // 获取common字段，如果不存在则创建
                            JSONObject common = jsonObj.getJSONObject("common");
                            if (common == null) {
                                common = new JSONObject();
                                jsonObj.put("common", common);
                            }

                            // 获取is_new字段，如果不存在则默认为"0"(老用户)
                            String isNew = common.getString("is_new");
                            if (isNew == null) {
                                isNew = "0";
                                common.put("is_new", isNew);
                            }

                            // 获取状态中存储的首次访问日期
                            String firstViewDt = firstViewDtState.value();
                            // 获取时间戳，如果不存在则使用当前时间
                            Long ts = jsonObj.getLong("ts");
                            if (ts == null) {
                                ts = System.currentTimeMillis();
                                jsonObj.put("ts", ts);
                            }
                            // 将时间戳转换为日期字符串
                            String dt = DateFormatUtil.toDate(ts);

                            // 新老用户识别逻辑
                            if ("1".equals(isNew)) {
                                // 如果是新用户
                                if (firstViewDt == null) {
                                    // 首次访问，更新状态
                                    firstViewDtState.update(dt);
                                } else if (!firstViewDt.equals(dt)) {
                                    // 如果首次访问日期与当前日期不同，则认为是老用户
                                    isNew = "0";
                                    common.put("is_new", isNew);
                                }
                            } else {
                                // 如果是老用户且没有首次访问日期，则设置为昨天
                                if (firstViewDt == null) {
                                    String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                                    firstViewDtState.update(yesterday);
                                }
                            }

                            // 输出处理后的数据
                            out.collect(jsonObj);
                        } catch (Exception e) {
                            // 打印处理异常
                            System.err.println("处理元素时出错: " + jsonObj.toJSONString());
                            e.printStackTrace();
                        }
                    }
                }
        );

        // 6. 定义侧输出流标签（用于分流）
        OutputTag<String> startTag = new OutputTag<String>("start") {};      // 启动日志流
        OutputTag<String> displayTag = new OutputTag<String>("display") {};  // 曝光日志流
        OutputTag<String> actionTag = new OutputTag<String>("action") {};    // 动作日志流
        OutputTag<String> errorTag = new OutputTag<String>("error") {};      // 错误日志流

        // 7. 分流处理
        SingleOutputStreamOperator<String> separatedStream = fixedStream.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context context, Collector<String> out) throws Exception {
                        try {
                            // 错误数据处理
                            JSONObject err = jsonObj.getJSONObject("err");
                            if (err != null) {
                                // 输出到错误流
                                context.output(errorTag, jsonObj.toJSONString());
                                // 移除err字段
                                jsonObj.remove("err");
                            }

                            // 启动数据处理
                            JSONObject start = jsonObj.getJSONObject("start");
                            if (start != null) {
                                // 输出到启动流
                                context.output(startTag, jsonObj.toJSONString());
                            } else {
                                // 页面数据处理
                                JSONObject page = jsonObj.getJSONObject("page");
                                if (page == null) {
                                    page = new JSONObject();
                                    jsonObj.put("page", page);
                                }

                                // 公共字段处理
                                JSONObject common = jsonObj.getJSONObject("common");
                                if (common == null) {
                                    common = new JSONObject();
                                    jsonObj.put("common", common);
                                }

                                // 时间戳处理
                                Long ts = jsonObj.getLong("ts");
                                if (ts == null) {
                                    ts = System.currentTimeMillis();
                                    jsonObj.put("ts", ts);
                                }

                                // 曝光数据处理
                                JSONArray displays = jsonObj.getJSONArray("displays");
                                if (displays != null && !displays.isEmpty()) {
                                    for (int i = 0; i < displays.size(); i++) {
                                        JSONObject display = displays.getJSONObject(i);
                                        if (display != null) {
                                            // 构建曝光日志对象
                                            JSONObject displayObj = new JSONObject();
                                            displayObj.put("display", display);
                                            displayObj.put("page", page);
                                            displayObj.put("common", common);
                                            displayObj.put("ts", ts);
                                            // 输出到曝光流
                                            context.output(displayTag, displayObj.toJSONString());
                                        }
                                    }
                                }

                                // 动作数据处理
                                JSONArray actions = jsonObj.getJSONArray("actions");
                                if (actions != null && !actions.isEmpty()) {
                                    for (int i = 0; i < actions.size(); i++) {
                                        JSONObject action = actions.getJSONObject(i);
                                        if (action != null) {
                                            // 构建动作日志对象
                                            JSONObject actionObj = new JSONObject();
                                            actionObj.put("action", action);
                                            actionObj.put("common", common);
                                            actionObj.put("page", page);
                                            actionObj.put("ts", ts);
                                            // 输出到动作流
                                            context.output(actionTag, actionObj.toJSONString());
                                        }
                                    }
                                }

                                // 移除已处理的字段
                                jsonObj.remove("displays");
                                jsonObj.remove("actions");

                                // 主流输出页面日志
                                out.collect(jsonObj.toJSONString());
                            }
                        } catch (Exception e) {
                            // 打印分流异常
                            System.err.println("分流处理时出错: " + jsonObj.toJSONString());
                            e.printStackTrace();
                            // 输出到错误流
                            context.output(errorTag, jsonObj.toJSONString());
                        }
                    }
                }
        );

        // 8. 获取各侧输出流
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);    // 启动日志流
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);// 曝光日志流
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);  // 动作日志流
        DataStream<String> errorDS = separatedStream.getSideOutput(errorTag);    // 错误日志流

        // 9. 定义输出主题
        String pageTopic = "topic_gd3_dwd_page";     // 页面日志主题
        String startTopic = "topic_gd3_dwd_start";    // 启动日志主题
        String displayTopic = "topic_gd3_dwd_display";// 曝光日志主题
        String actionTopic = "topic_gd3_dwd_action";  // 动作日志主题
        String errorTopic = "topic_gd3_dwd_error";    // 错误日志主题

        // 10. 打印各流数据（调试用）
        separatedStream.print("page>>>");      // 页面日志
        startDS.print("start!!!");             // 启动日志
        displayDS.print("display@@@");         // 曝光日志
        actionDS.print("action###");           // 动作日志
        errorDS.print("error$$$");             // 错误日志

        // 11. 写入Kafka
        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(pageTopic));    // 页面日志
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(startTopic));          // 启动日志
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(displayTopic));      // 曝光日志
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(actionTopic));        // 动作日志
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(errorTopic));          // 错误日志

        // 12. 执行作业
        env.execute("DwdGd3LogApp");
    }
}
